package com.atguigu.source

import java.util.Calendar

import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext

import scala.util.Random


//泛型是SensorReasing
class SensorSource extends RichParallelSourceFunction[SensorReading]{
  //表示数据源是否正常运行
  var running:Boolean=true

    //上下文参数用来发出数据
  override def run(ctx:SourceContext[SensorReading]): Unit = {
      val rand=new Random()

      var curFTemp= (1 to 10).map(
         //使用高斯噪声产生随机温度值
        i=>("sensor_"+i,(rand.nextGaussian()*20))
      )

      while (running){
        curFTemp =curFTemp.map(
          t=>(t._1,t._2+(rand.nextGaussian()*0.5))
        )
        val curTime = Calendar.getInstance().getTimeInMillis
        //产生ms为单位的时间戳
        curFTemp.foreach(


          t=>(ctx.collect(SensorReading(t._1,curTime,t._2))))


          Thread.sleep(300)



      }




  }

  override def cancel(): Unit = false

}
